package homework1

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object KafkaDStream {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("KafkaDStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    val groupId = "group1"
    val topics = Array("topic1")
    val kafkaParams = getKafkaConsumerParameters(groupId)

    val dsSteam = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    // 先将读取文件格式进行转化
    val newStream = dsSteam.map(message => {
      val key = message.key()
      val value = message.value()
      val newValue = value.replace(",", "|").replace("<<<!>>>", "")
      (key, newValue)
    })
    newStream.print(10)

    //将转化后的数据发到kafka
    newStream.foreachRDD(rdd => {
      rdd.foreachPartition(it =>{
        val kafkaProducer = KafkaProducer.instance()

        if (it.hasNext) {
          val message = new ProducerRecord[String, String](KafkaProducer.topic, it.next()._1, it.next()._2)
          kafkaProducer.send(message)
        }
      })
    })

  }

  def getKafkaConsumerParameters(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092,linux122:9092,linux123:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid
      //ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
      //      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }

}
